package com.easy.test.rocket;

import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;  
  
/** 
 * Created by zkn on 2016/10/30. 
 */  
public class ConsumerTest02{  
  
    public static void main(String[] args) {  
  
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("doooly-message1");  
        //pushConsumer.setNamesrvAddr("192.168.180.1:9876");  
        pushConsumer.setNamesrvAddr("39.98.195.15:9876");  
        pushConsumer.setInstanceName("Producer");  
        /** 
         * 订阅指定topic下所有消息<br> 
         * 注意：一个consumer对象可以订阅多个topic 
         */  
        try {  
            pushConsumer.subscribe("test_topic", "*");  
            pushConsumer.registerMessageListener(new MessageListenerConcurrently() {  
                 @Override  
                  public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {  
                        MessageExt messageExt = msgs.get(0);  
                         System.out.println(new String(messageExt.getBody()));  
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
                 }  
            }  
            );  
        } catch (MQClientException e) {  
            e.printStackTrace();  
        }  
        try {  
            pushConsumer.start();  
        } catch (MQClientException e) {  
            e.printStackTrace();  
        }  
    }  
}  
